package raft.service.impl;

import raft.entity.*;
import raft.node.NodeStatus;
import raft.service.Consensus;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class ConsensusImpl implements Consensus {

    public NodeImpl node;

    public final ReentrantLock voteLock = new ReentrantLock();
    public final ReentrantLock appendLock = new ReentrantLock();

    public ConsensusImpl(NodeImpl node){
        this.node = node;
    }

    /**
     * 注意，选举操作应该是串行的，因为涉及到状态修改，并发操作将导致数据错乱。也就是说，如果抢锁失败，应当立即返回错误。
     * 首先判断对方的 term 是否小于自己，如果小于自己，直接返回失败。
     * 如果当前节点没有投票给任何人，或者投的正好是对方，那么就可以比较日志的大小，反之，返回失败。
     * 如果对方日志没有自己大，返回失败。反之，投票给对方，并变成 follower。变成 follower 的同时，
     * 异步的选举任务在最后从 condidate 变成 leader 之前，会判断是否是 follower，如果是 follower，就放弃成为 leader。这是一个兜底的措施。
     * @param param
     * @return
     */

    @Override
    public RvoteResult requestVote(RvoteParam param) {

        try {
            RvoteResult.Builder builder = RvoteResult.newBuilder();
            //没获得锁
            if(!voteLock.tryLock()){
                return builder.term(node.getCurrentTerm()).vote(false).build();
            }
            System.out.println("....选举");
            //任期低于自己
            if(param.getTerm()< node.getCurrentTerm()){
                return builder.term(node.getCurrentTerm()).vote(false).build();
            }

            if ( node.getVotedFor()==null ||node.getVotedFor().equals("")||node.getVotedFor().equals(param.getCandidateId())  ){
                 if(node.getLogModule().getLast()!=null){
                     if(node.getLogModule().getLast().getTerm()> param.getTerm()){
                         return builder.vote(false).build();
                     }

                     if(node.getLogModule().getLastIndex()>param.getLastLogIndex()){
                         return builder.vote(false).build();
                     }
                 }

                node.status = NodeStatus.FOLLOWER;

                node.peerSet.setLeader(param.getCandidateId());
                node.setCurrentTerm(param.getTerm());
                node.setVotedFor(param.getCandidateId());
                return builder.term(node.currentTerm).vote(true).build();

            }

            return builder.term(node.currentTerm).vote(false).build();

        }finally {
            System.out.println("....选举。。。。jiesu ");
            voteLock.unlock();
        }

    }

    /**
     * 附加日志(多个日志,为了提高效率) RPC
     *
     * 接收者实现：
     *    如果 term < currentTerm 就返回 false （5.1 节）
     *    如果日志在 prevLogIndex 位置处的日志条目的任期号和 prevLogTerm 不匹配，则返回 false （5.3 节）
     *    如果已经存在的日志条目和新的产生冲突（索引值相同但是任期号不同），删除这一条和之后所有的 （5.3 节）
     *    附加任何在已有的日志中不存在的条目
     *    如果 leaderCommit > commitIndex，令 commitIndex 等于 leaderCommit 和 新日志条目索引值中较小的一个
     */
    @Override
    public AentryResult appendEntries(AentryParam param) {

     try {
         if(!appendLock.tryLock()){return new AentryResult(false);}
         //心跳
         /**
          * 更新心跳、选举时间
          * 状态设置为选举
          * 申明领导
          */
         AentryResult res = new AentryResult(node.currentTerm);
         if(param.getTerm()>=node.currentTerm){
             node.status = NodeStatus.FOLLOWER;
             node.setCurrentTerm(param.getTerm());
             node.setPreHeartBeat(System.currentTimeMillis());
             node.setPreElection(System.currentTimeMillis());
             node.peerSet.setLeader(param.getLeaderId());
             res.setSuccess(true);
         }else{
             res.setSuccess(false);
         }

         if(param.getEntries() == null|| param.getEntries().length == 0){
             return  res;
         }

         //不是心跳  日志附加

         //如果小于当前节点任期
         if(param.getPreLogTerm()<node.currentTerm||node.logModule.read(param.getPrevLogIndex()).getTerm()!= param.getTerm()){
             return AentryResult.fail();
         }

         //日志起冲突
         /**
          *如果存在要插入的节点 且 节点的任期与 param要插入的节点 任期冲突   就删除node的日志参数
          *如果不存在任期冲突  就返回   因为有数据就不插入 了
          *
          */
         LogEntry ExistlogEntry = node.logModule.read(param.getPrevLogIndex()+1);
         if(ExistlogEntry!=null){

           if(ExistlogEntry.getTerm() != param.getEntries()[0].getTerm()){
               node.logModule.removeOnStarIndex(param.getPrevLogIndex()+1);
           }else {
               res.setSuccess(true);
               return res;
           }

         }

         //添加到日志 状态机
         for(LogEntry logEntry : param.getEntries()){
             node.logModule.write(logEntry);
             node.stateMachine.apply(logEntry);
         }

         /**
          * 取最小的的日志提交记录 作为 当前节点的提交记录
          */
         if(param.getLeaderCommitIndex() > node.commitIndex ){
             int finalindex = (int)Math.min(node.commitIndex,node.logModule.getLastIndex());
             node.setCommitIndex(finalindex);
             node.setLastApplied(finalindex);
         }

         node.status = NodeStatus.FOLLOWER;
         res.setSuccess(true);

         return res;
     }finally {
         System.out.println("Beat...");
         appendLock.unlock();
     }

    }


}
